package cn.flkj.evidence.RocketMQConsumer;

import cn.flkj.evidence.EvidenceImpl;
import cn.flkj.evidence.RocketMQConsumer.Utils.EsUtil;
import cn.flkj.evidence.RocketMQConsumer.Utils.OperationTxt;
import cn.flkj.evidence.repositories.EvidenceMapper;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.log4j.PropertyConfigurator;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.consumer.PullResultExt;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.*;

@Component
public class RocketMQConsumer extends MessageExt {
    private EvidenceImpl impl;
    private EvidenceMapper evidenceMapper;

    private OperationTxt offsetValueFile;
    private String index;
    protected final static Logger logger = LoggerFactory.getLogger(RocketMQConsumer.class);
    private static final Map<MessageQueue, Long> offseTable = new HashMap<>();
    private RestHighLevelClient client;
    private String namesrvAddr;
    private String topic;
    private String consumerGroup;
    private String tag;
    private Integer pullBatchSize;
    private Map<String, Map<String, JSONObject>> messageBatch = new HashMap<>();
    private Map<String, Long> indexTime = new HashMap<>();
    private SentToEvidence sentToEvidence = new SentToEvidence();

    //----------------------------------------------------------------------------


    public RocketMQConsumer(EvidenceImpl impl, EvidenceMapper evidenceMapper) {
        this.impl = impl;
        this.evidenceMapper = evidenceMapper;
    }
    public RocketMQConsumer() {
    }

        @Autowired
    private  EvidenceImpl evidence;
    //@Resource
    //private EvidenceMapper evidenceMapper;
//    @Autowired
//    private RocketMQToEvidence rocketMQToEvidence;
    //----------------------------------------------------------------------------
//    public RocketMQToFL(){
//        try {
//            this.impl = new EvidenceImpl();
//        }catch (Exception e){
//
//        }
//
//    }


    public void init() {
        Properties properties = new Properties();
        FileInputStream arg = null;
        try {
            arg = new FileInputStream(System.getProperty("user.dir") + "/config/application.properties");
        } catch (FileNotFoundException e) {
            logger.error(e.getMessage());
        }
        try {
            properties.load(arg);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
        topic = properties.getProperty("ROCKETMQ_TOPIC");
        namesrvAddr = properties.getProperty("ROCKETMQ_NAMESRV");
        consumerGroup = properties.getProperty("ROCKETMQ_COSUMER_GROUP_TO_FUCHAIN");
        tag = properties.getProperty("ROCKETMQ_TAG");
        pullBatchSize = Integer.parseInt(properties.getProperty("ROCKETMQ_PULL_BATCH_SIZE"));
        Map<String, String> propertiesMap = new HashMap<>();
        propertiesMap.put("hosts", properties.getProperty("ES_HOSTS"));
        propertiesMap.put("username", properties.getProperty("ES_USERNAME"));
        propertiesMap.put("password", properties.getProperty("ES_PASSWORD"));
        client = new HighLevelClient(propertiesMap).getClient();
        start();
    }

    private void saveToEvident(Map<String, Map<String, JSONObject>> messageBatch,String index) {
        FileInputStream fileInputStream = null;
        try {
            fileInputStream = new FileInputStream(System.getProperty("user.dir") + "/config/application.properties");
        } catch (FileNotFoundException e) {
            logger.error(e.getMessage());
        }
        Properties properties = new Properties();
        try {
            properties.load(fileInputStream);
        } catch (IOException e) {
            logger.error(e.getMessage());
        }
//        String keyTail = properties.getProperty("KEY_TAIL");


            for (String id : messageBatch.get(index).keySet()) {
                JSONObject result = null;
                JSONObject temp = messageBatch.get(index).get(id);
                String dataHash = temp.getJSONObject("head").getString("dataHash");
                try {
                    result = sentToEvidence.send(dataHash,impl,evidenceMapper);
                }catch (Exception e){

                }
                int code = result.getInteger("code");
                if (code == 200) {
                    logger.info("上链结果：" + result);
                    Set<String> idSet = new HashSet<>();
                    idSet.add(id);
                    String evidenceAddress = result.getString("evidenceAddress");
                    EsUtil.BulEvidenceAddress(client, idSet, index, evidenceAddress);
                    logger.info("evidenceAddress",evidenceAddress);
                }

            }

        }



    public void start() {
        // 读取 日志 配置文件
        String log4j = System.getProperty("user.dir") + "/config/log4j.properties";
        PropertyConfigurator.configure(log4j);

        offsetValueFile=new OperationTxt("offsetValue.txt");


        while (true) {
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
            consumer.setNamesrvAddr(namesrvAddr);
            try {
                consumer.start();
                Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);
                for (MessageQueue mq : mqs) {
                    logger.info("Consume(RocketMQToFL) from the queue: " + mq);

                    // 从持久化文件中获取当前 offset 列表
                    Long offset;
                    JSONObject offsetList = JSON.parseObject(offsetValueFile.getContent());
                    if (offsetList.getString(mq.toString()) == null) {
                        offset = 0L;
                    } else {
                        offset = Long.valueOf(offsetList.getString(mq.toString()));
                    }
                    logger.info("当前消费位点为 " + offset);

                    PullResultExt pullResult = (PullResultExt) consumer.pullBlockIfNotFound(mq, tag, offset, pullBatchSize);
                    offset = pullResult.getNextBeginOffset();
                    logger.info("消费位点更新为 " + offset);
                    offsetList.put(mq.toString(), offset);
                    // 将新的 offsetList 写入持久化文件
                    offsetValueFile.setContent(offsetList.toJSONString());

                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                            Map<String, JSONObject> map;
                            for (MessageExt msg : messageExtList) {
                                map = new HashMap<>();
                                String message = new String(msg.getBody());
                                JSONObject info = JSON.parseObject(message);
                                JSONObject head = info.getJSONObject("head");
                                JSONObject body = info.getJSONObject("body");
                                index = head.getString("dataUser");
                                String id = head.getString("dataHash");
                                map.put(id, info);
                                //存数据
                                messageBatch = saveData(index, messageBatch, map);
                            }
                            break;
                        case NO_MATCHED_MSG:
                            break;
                        case NO_NEW_MSG:
                            break;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                    }
                    long currentTime = System.currentTimeMillis();
                    List<String> indexList = new LinkedList<>();
                    if (!messageBatch.isEmpty()) {
                        for (String index : messageBatch.keySet()) {
                            //数据缓存的时间（单位：分钟）
                            long time = (currentTime - indexTime.get(index)) / 1000 / 60;
                            //数据缓存的数量
                            long size = messageBatch.get(index).size();
                            if (time > 3 || size > 0) {
                                //发送数据
                                saveToEvident(messageBatch,index);
                                indexList.add(index);
                            }
                        }
                    }
                    //remove
                    if (indexList.size() > 0) {
                        for (int n = 0; n < indexList.size(); n++) {
                            String index = indexList.get(n);
                            messageBatch.remove(index);
                            indexTime.remove(index);
                        }
                    }

                }
            } catch (InterruptedException e) {
                logger.error("InterruptedException:", e);
            } catch (RemotingException e) {
                logger.error("RemotingException:", e);
            } catch (MQClientException e) {
                logger.error("MQClientException:", e);
            } catch (MQBrokerException e) {
                logger.error("MQBrokerException:", e);
            } finally {
                consumer.shutdown();
            }


        }
    }

//    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
//        offseTable.put(mq, offset);
//    }
//
//    private static long getMessageQueueOffset(MessageQueue mq) {
//        Long offset = offseTable.get(mq);
//        if (offset != null)
//            return offset;
//        return 0;
//    }

    //存数据
    public Map<String, Map<String, JSONObject>> saveData(String index, Map<String, Map<String, JSONObject>> messageBatch, Map<String, JSONObject> newMap) {
        if (messageBatch.containsKey(index)) {
            Map<String, JSONObject> combineResultMap = new HashMap();
            Map<String, JSONObject> map = messageBatch.get(index);
            combineResultMap.putAll(map);
            combineResultMap.putAll(newMap);
            //更新index对应的map
            messageBatch.put(index, combineResultMap);
        } else {
            messageBatch.put(index, newMap);
            indexTime.put(index, System.currentTimeMillis());
        }
        return messageBatch;
    }



}
